﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Data;
using System.Data.SqlClient;
using KingSite.SyncDB.Core.Entity;
using KingSite.SyncDB.Core.Repository;
using Microsoft.Practices.EnterpriseLibrary.Data.Sql;
using KingSite.SyncDB.Repository;
using System.Collections;


namespace KingSite.SyncDB.Repository {
    public class MSSqlDataAccess : AbstractDataAccess {

        public MSSqlDataAccess(SyncJobInfo job, IList<SyncEntity> syncList)
            : base(job, syncList) {

        }

        public override DataSet GetSourceTable(string lastTimeStamp, int startRow, int endRow) {
            StringBuilder sb = new StringBuilder();
            sb.Append("select ");
            string columns = GetSelectColumn();
            sb.Append(columns);
            sb.Append(" from ");
            sb.Append(string.Format(" ( SELECT ROW_NUMBER() over(order by {0}) row , ", GetListSplit(primaryKeys)));
            sb.Append(columns);
            sb.Append(" from ");
            sb.Append(job.STableName);
            sb.Append(" with(readpast) where  [timestamp] >  ");
            sb.Append(lastTimeStamp);
            sb.Append(" ) t");
            sb.Append(string.Format(" where  t.row between {0} and {1} ", startRow, endRow));
            string sql = sb.ToString();
            //这里要连接源数据库查询
            DataSet ds = null;
            try {
                SqlDatabase sdb = new SqlDatabase(job.SDbconnstring);
                ds = sdb.ExecuteDataSet(CommandType.Text, sql);
            }
            catch (Exception ex) {
                log.Info(string.Format("任务{0}发生错误：连接源数据库失败! {1}", job.JobName, ex.Message));
            }

            return ds;
        }

        private string GetSelectColumn() {
            StringBuilder sb = new StringBuilder();
            foreach (string key in primaryKeys) {
                sb.Append(key); //源表一定要带有 主键列 ， timestamp 字段
                sb.Append(", ");
            }
            sb.Append(" [timestamp] ");
            foreach (SyncEntity info in syncList) {
                //排除主键值
                if (!info.SCIsPrimaryKey) {
                    sb.Append(",");
                    sb.Append(info.SColumnName);
                }
            }
            return sb.ToString();
        }

        private string GetListSplit(List<string> list) {
            StringBuilder sb = new StringBuilder();
            foreach (string key in list) {
                sb.Append(key); //源表一定要带有 主键列 ， timestamp 字段
                sb.Append(", ");
            }
            sb.Remove(sb.Length - 2, 2);
            return sb.ToString();
        }

        public override List<ExecSqlObject> GetUpdateSql() {
            List<ExecSqlObject> result = null;
            if (updatePrimaryKeyValues != null) {
                result = new List<ExecSqlObject>();
                foreach (Hashtable ht in updatePrimaryKeyValues) {
                    StringBuilder sb = new StringBuilder();
                    sb.Append("update ");
                    sb.Append(job.DTableName);
                    sb.Append(" set ");
                    List<SqlParameter> paras = new List<SqlParameter>();
                    foreach (SyncEntity info in syncList) {
                        if (!info.SCIsAutoIdentity && !info.SCIsPrimaryKey) { //自增列、主键 不要更新
                            sb.Append(string.Format("{0} = @{0},", info.DColumnName));
                            object value = GetSourceColumnValue(ht, info.DColumnName);
                            paras.Add(new SqlParameter(info.DColumnName, value));
                        }
                    }
                    sb.Remove(sb.Length - 1, 1); //移除最后一个 ,
                    sb.Append(" where ( ");
                    foreach (DictionaryEntry kv in ht) {
                        sb.Append(string.Format("{0} = @{0}", kv.Key));
                        sb.Append(" and ");
                        SqlParameter para = new SqlParameter(Convert.ToString(kv.Key), kv.Value);
                        paras.Add(para);
                    }
                    sb.Remove(sb.Length - 5, 5); //移除最后一个 and
                    sb.Append(" )");

                    object tmp = GetRowTimeStamp(ht);
                    string rowTimestamp = ConvertToStringFromBytes(tmp);
                    result.Add(new ExecSqlObject { KeyValues = ht, Sql = sb.ToString(), ParaList = paras, RowTimeStamp = rowTimestamp });
                }
            }
            return result;
        }

        public override List<ExecSqlObject> GetInsertSql() {
            List<ExecSqlObject> result = null;
            if (insertPrimaryKeyValues != null) {
                result = new List<ExecSqlObject>();
                foreach (Hashtable ht in insertPrimaryKeyValues) {
                    StringBuilder sb = new StringBuilder();
                    sb.Append("insert into ");
                    sb.Append(job.DTableName);
                    sb.Append(" ( ");
                    List<SqlParameter> paras = new List<SqlParameter>();
                    StringBuilder tmpstr = new StringBuilder();
                    foreach (SyncEntity info in syncList) {
                        if (!info.DCIsAutoIdentity) { //目标表的自增列不要插入
                            sb.Append(string.Format(" {0}, ", info.DColumnName));
                            object value = GetSourceColumnValue(ht, info.DColumnName);
                            paras.Add(new SqlParameter(info.DColumnName, value));
                            tmpstr.Append(string.Format("@{0},", info.DColumnName));
                        }
                    }
                    tmpstr.Remove(tmpstr.Length - 1, 1);
                    sb.Remove(sb.Length - 2, 2); //移除最后一个 ,
                    sb.Append(" ) values (");
                    sb.Append(tmpstr.ToString());
                    sb.Append(" )");
                    object tmp = GetRowTimeStamp(ht);
                    string rowTimestamp = ConvertToStringFromBytes(tmp);
                    result.Add(new ExecSqlObject { KeyValues = ht, Sql = sb.ToString(), ParaList = paras, RowTimeStamp = rowTimestamp });
                }
            }
            return result;
        }

        public override void ExecSyncSql(SyncJobDetailInfo detail) {
            SqlDatabase sdb = new SqlDatabase(job.DDbconnstring);

            if (job.IsSyncUpdate && detail.UpdateExecList != null) {
                foreach (ExecSqlObject so in detail.UpdateExecList) {
                    ExecSqlDo(sdb, so, "update");
                }
            }

            if (job.IsSyncInsert && detail.InsertExecList != null) {
                foreach (ExecSqlObject so in detail.InsertExecList) {
                    ExecSqlDo(sdb, so, "insert");
                }
            }
        }

        private void ExecSqlDo(SqlDatabase sdb, ExecSqlObject so, string optype) {
            try {
                SqlCommand comm = new SqlCommand();
                comm.CommandText = so.Sql;
                comm.Parameters.AddRange(so.ParaList.ToArray());
                sdb.ExecuteNonQuery(comm);
                UpdateRowTimeStamp(so);
                job.SLastTimeStamp = so.RowTimeStamp;
                AddLog(so, optype);
            }
            catch (Exception ex) {
                log.Info("error : " + ex.Message);
            }
        }

        public override SyncJobDetailInfo GetSyncJobDetailInfo() {
            SyncJobDetailInfo result = new SyncJobDetailInfo();
            if (job.IsSyncUpdate) {
                result.UpdateExecList = GetUpdateSql();
            }
            if (job.IsSyncInsert) {
                result.InsertExecList = GetInsertSql();
            }
            return result;
        }

        public override int GetTotalRecords() {
            int count = 0;
            StringBuilder sb = new StringBuilder();
            sb.Append("select count(1) from ");
            sb.Append(job.STableName);
            sb.Append(" where [timestamp] > ");
            sb.Append(job.SLastTimeStamp);
            string sql = sb.ToString();
            //这里要连接源数据库查询            
            try {
                SqlDatabase sdb = new SqlDatabase(job.SDbconnstring);
                DataSet ds = sdb.ExecuteDataSet(CommandType.Text, sql);
                if (ds != null && ds.Tables[0].Rows.Count > 0) {
                    count = Convert.ToInt32(ds.Tables[0].Rows[0][0]);
                }
            }
            catch (Exception ex) {
                log.Info("连接源数据库失败!" + ex.Message);
            }

            return count;
        }

        public override string GetTableMaxTimeStamp(string dbConnstring, string tableName) {
            string result = "0x0000000000000001";
            string sql = string.Format(@"select top 1 [timestamp] from {0} order by [timestamp] desc", tableName);
            SqlDatabase sdb = new SqlDatabase(dbConnstring);
            DataSet ds = sdb.ExecuteDataSet(CommandType.Text, sql);
            if (ds != null && ds.Tables[0].Rows.Count > 0) {
                result = ConvertToStringFromBytes(ds.Tables[0].Rows[0][0]);
            }
            return result;
        }

        private static string ConvertToStringFromBytes(object obj) {
            byte[] dd = (byte[])obj;
            string result = "0x" + BitConverter.ToString(dd).Replace("-", "");
            return result;
        }
    }
}
